package com.shujia.redis;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.Jedis;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Set;

public class HBaseIOnRedisIndex {

    private HConnection connection;
    private HBaseAdmin hBaseAdmin;
    Jedis jedis;

    @Before
    public void init() {

        //创建hadoop配置文件对象
        Configuration configuration = new Configuration();

        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");


        try {
            //建立连接   和zk的连接
            connection = HConnectionManager.createConnection(configuration);

            hBaseAdmin = new HBaseAdmin(configuration);

            jedis = new Jedis("node3", 6379);

            System.out.println("连接建立成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() throws Exception {
        connection.close();
        jedis.close();
    }


    @Test
    public void putALL() throws Exception {


        //create 'student','info'

        HTableInterface student = connection.getTable("student");

        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {

            String[] split = line.split(",");

            String id = split[0];
            String name = split[1];
            Integer age = Integer.parseInt(split[2]);
            String gender = split[3];
            String clazz = split[4];


            Put put = new Put(id.getBytes());
            put.add("info".getBytes(), "name".getBytes(), name.getBytes());
            put.add("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));
            put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());


            //保存数据
            student.put(put);

            //保存索引
            //使用redis set 集合保存行与列值的映射关系
            jedis.sadd("student_index_gender:" + gender, id);

        }

    }

    @Test
    public void query() throws Exception {

        String gender = "男";

        //1、查询索引表
        //2、查询数据表

        long l = System.currentTimeMillis();

        Set<String> ids = jedis.smembers("student_index_gender:" + gender);

        long l1 = System.currentTimeMillis();

        System.out.println(l1 - l);

        ArrayList<Get> gets = new ArrayList<>();


        for (String id : ids) {
            Get get = new Get(id.getBytes());
            gets.add(get);
        }


        HTableInterface student = connection.getTable("student");

        Result[] results = student.get(gets);


        for (Result result : results) {

            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String gender1 = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            Integer age = Bytes.toInt(result.getValue("info".getBytes(), "age".getBytes()));

            System.out.println(id + "\t" + name + "\t" + gender1 + "\t" + clazz + "\t" + age);
        }


    }


}
